package com.datasync.core;

import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;

import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
public class CdcListener implements ApplicationListener<ContextRefreshedEvent> {
    
    private final DataSourceConfig dataSourceConfig;
    private final ExecutorService executor = Executors.newSingleThreadExecutor();

    public CdcListener(DataSourceConfig dataSourceConfig) {
        this.dataSourceConfig = dataSourceConfig;
    }

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        Properties props = new Properties();
        props.setProperty("name", "datasync-engine");
        props.setProperty("connector.class", getConnectorClass());
        props.setProperty("database.hostname", dataSourceConfig.getHost());
        props.setProperty("database.port", String.valueOf(dataSourceConfig.getPort()));
        props.setProperty("database.user", dataSourceConfig.getUsername());
        props.setProperty("database.password", dataSourceConfig.getPassword());
        props.setProperty("database.dbname", dataSourceConfig.getDatabase());
        props.setProperty("database.server.name", "datasync_server");

        DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
                .using(props)
                .notifying(this::handleEvent)
                .build();

        executor.execute(engine);
    }

    private String getConnectorClass() {
        return switch (dataSourceConfig.getType()) {
            case GAUSSDB -> "com.huawei.gaussdb.connect.GaussdbConnector";
            case TDSQL -> "com.tencent.tdsql.connect.TdsqlConnector";
            default -> throw new IllegalArgumentException("Unsupported database type");
        };
    }

    private void handleEvent(ChangeEvent<String, String> event) {
        String key = event.key();
        String value = event.value();
        log.info("捕获到数据变更事件 - Key: {}, Value: {}", key, value);
        // 触发增量同步任务
syncService.triggerIncrementalSync(event.value());
    }
}